package org.trade.common.rocketmq;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.trade.common.constant.MQEnums;
import org.trade.common.exception.AceMQException;


/**
 * @author fangfeiyue
 *
 */
public class AceMQProducer {

	public static final Logger logger = LoggerFactory.getLogger(AceMQProducer.class);

	private DefaultMQProducer producer;

	private String groupName;
	private String namesrvAddr;
	// 单条消息的最大值，4M
	private int maxMessageSize = 1024 * 1024 * 4;
	// 时间延迟时间，10秒
	private int sendMsgTimeout = 10000;

	public AceMQProducer() {

	}

	/**
	 * 初始化
	 */
	public void init() throws AceMQException {
		if (StringUtils.isBlank(this.groupName)) {
			throw new AceMQException("groupname is blank");
		}
		if (StringUtils.isBlank(this.namesrvAddr)) {
			throw new AceMQException("namesrvAddr is blank");
		}
		this.producer = new DefaultMQProducer(this.groupName);
		this.producer.setNamesrvAddr(this.namesrvAddr);
		this.producer.setMaxMessageSize(maxMessageSize);
		this.producer.setSendMsgTimeout(sendMsgTimeout);
		try {
			this.producer.start();
			logger.info(String.format("producer is start!groupName:{%s},namesrvAddr:{%s}", this.groupName,
					this.namesrvAddr));
		} catch (MQClientException e) {
			e.printStackTrace();
			throw new AceMQException(e);
		}
	}

	public void setProducer(DefaultMQProducer producer) {
		this.producer = producer;
	}

	public void setGroupName(String groupName) {
		this.groupName = groupName;
	}

	public void setNamesrvAddr(String namesrvAddr) {
		this.namesrvAddr = namesrvAddr;
	}

	public void setMaxMessageSize(int maxMessageSize) {
		this.maxMessageSize = maxMessageSize;
	}

	public void setSendMsgTimeout(int sendMsgTimeout) {
		this.sendMsgTimeout = sendMsgTimeout;
	}

	/*
	 * 发送信息
	 */
	public SendResult sendMessage(String topic, String tags, String keys, String messageText) throws AceMQException {
		Message message = new Message(topic, tags, keys, messageText.getBytes());
		SendResult sendResult = null;
		try {
			sendResult = this.producer.send(message);
		} catch (Exception e) {
			logger.error("send message error:{}", e.getMessage(), e);
			throw new AceMQException(e);
		}
		return sendResult;
	}

	public SendResult sendMessage(MQEnums.TopicEnum topicEnum, String keys, String messageText) throws AceMQException {
		return this.sendMessage(topicEnum.getTopic(), topicEnum.getTag(), keys, messageText);
	}
}
